package com.citusdata.elven.httpRpc;

import com.citusdata.elven.httpRpc.HttpRpcUrl;
import com.google.protobuf.Descriptors.ServiceDescriptor;
import com.google.protobuf.Descriptors.MethodDescriptor;
import com.google.protobuf.Message;
import com.google.protobuf.RpcCallback;
import com.google.protobuf.Service;
import com.google.protobuf.UninitializedMessageException;
import java.io.IOException;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;

import org.apache.http.*;
import org.apache.http.entity.*;
import org.apache.http.protocol.*;
import org.apache.http.util.*;
import org.apache.log4j.Logger;


class HttpRpcRequestHandler implements HttpRequestHandler {
  private static final Logger logger = Logger.getLogger(HttpRpcRequestHandler.class);

  private final ConcurrentMap<String, Service> serviceRegistry;

  public HttpRpcRequestHandler(ConcurrentMap<String, Service> serviceRegistry) {
    this.serviceRegistry = serviceRegistry;
  }


  public void handle(HttpRequest httpRequest, HttpResponse httpResponse, 
                     HttpContext httpContext) {

    // Resolve the service and method names
    String urlString = httpRequest.getRequestLine().getUri();
    HttpRpcUrl requestUrl = new HttpRpcUrl(urlString);
    if (!requestUrl.valid()) {
      handleError(httpResponse, "Invalid URL: " + requestUrl.getUrl(),
                  HttpStatus.SC_BAD_REQUEST, null);
      return;
    }

    String serviceName = requestUrl.getServiceName();
    Service service = serviceRegistry.get(serviceName);
    if (service == null) {
      handleError(httpResponse, "Could not find service: " + serviceName,
                  HttpStatus.SC_NOT_IMPLEMENTED, null);
      return;
    }

    String methodName = requestUrl.getMethodName();
    ServiceDescriptor descriptor = service.getDescriptorForType();
    MethodDescriptor method = descriptor.findMethodByName(methodName);
    if (method == null) {
      handleError(httpResponse, "Could not find method: " + methodName,
                  HttpStatus.SC_NOT_IMPLEMENTED, null);
      return;
    }

    // Get the request entity
    if (!(httpRequest instanceof HttpEntityEnclosingRequest)) {
      handleError(httpResponse, "Malformed request object", 
                  HttpStatus.SC_BAD_REQUEST, null);
      return;
    }
    HttpEntityEnclosingRequest enclosing = (HttpEntityEnclosingRequest) httpRequest;

    Message request = null;
    try {
      byte[] requestBytes = EntityUtils.toByteArray(enclosing.getEntity());
      Message requestPrototype = service.getRequestPrototype(method);
      Message.Builder requestBuilder = requestPrototype.newBuilderForType();
      requestBuilder.mergeFrom(requestBytes);
      request = requestBuilder.build();
    } catch (IOException ex) {
      handleError(httpResponse, "Invalid request protocol",
                  HttpStatus.SC_UNPROCESSABLE_ENTITY, ex);
      return;
    } catch (UninitializedMessageException ex) {
      handleError(httpResponse, "Request protocol missing required fields",
                  HttpStatus.SC_UNPROCESSABLE_ENTITY, ex);
      return;
    }

    // Call the method
    HttpRpcController httpController = new HttpRpcController();
    Callback callback = new Callback();
    service.callMethod(method, httpController, request, callback);

    // Check if the call succeeded
    if (httpController.failed()) {
      handleError(httpResponse, "RPC call failed", 
                  HttpStatus.SC_INTERNAL_SERVER_ERROR, null);
      return;
    }

    // The call succeeded, set the response and return
    httpResponse.setStatusCode(HttpStatus.SC_OK);
    Message response = callback.getResponse();
    if (response != null) {
      httpResponse.addHeader("Content-Type", "application/octet-stream");
      httpResponse.setEntity(new ByteArrayEntity(response.toByteArray()));
    }
  }


  private void handleError(HttpResponse response,
                           String message, int statusCode, Exception ex) {
    logger.warn(message + ": " + statusCode + " failure with exception: " + ex);
    response.setStatusCode(statusCode);
  }

  // Inner class Callback that just saves the response and the fact that the
  // function was invoked.
  private class Callback implements RpcCallback {
    private Message response;
    
    public void run(Object response) {
      this.response = (Message) response;
    }

    public Message getResponse() {
      return response;
    }
  }

}